package cn.tonyandmoney.baidu.bus

import cn.tonyandmoney.baidu.consumers.RxConsumer
import io.reactivex.Observable
import io.reactivex.functions.Consumer
import io.reactivex.schedulers.Schedulers
import io.reactivex.subjects.PublishSubject

/**
 * 这个是什么情况
 */
class SpeechBus {
    val mBus = PublishSubject.create<Any>()

    fun createErrorConsumer(): Consumer<Throwable> {
        return Consumer { it.printStackTrace() }
    }


    companion object {

        val BUS = SpeechBus()

        fun <T> register(clazz: Class<T>): Observable<T> {
            return BUS.mBus.ofType(clazz)
                    .subscribeOn(Schedulers.io())
                    .unsubscribeOn(Schedulers.io())
        }

        inline fun <reified T> register(): Observable<T> {
            return BUS.mBus.ofType(T::class.java)
        }

        inline fun <reified T> register(consumer: Consumer<T>) {
            BUS.mBus.ofType(T::class.java)
                    .subscribeOn(Schedulers.computation())
                    .unsubscribeOn(Schedulers.computation())
                    .subscribe(RxConsumer(consumer), BUS.createErrorConsumer())
        }

        fun post(obj: Any) = BUS.mBus.onNext(obj)
    }
}